package data.platform.common.stream.aggregates;

import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

import java.time.LocalDateTime;
import java.util.concurrent.atomic.AtomicBoolean;

@Getter
@Setter
@NoArgsConstructor
public class WindowAggregateItem {

    private Double avg;

    private Double max;

    private Double min;

    private Double sum;

    private Integer count;

    private Double latestValue;

    private LocalDateTime latestEventTime;

    @JsonIgnore
    private AtomicBoolean dirty = new AtomicBoolean(true);

    @JsonIgnore
    private static final String KEY_PRE_NAME = "stream";

    @JsonIgnore
    public static final String KEY_SEPARATOR = "|";

    public WindowAggregateItem(Double avg, Double max, Double min, Double sum, Integer count) {
        this.avg = avg;
        this.max = max;
        this.min = min;
        this.sum = sum;
        this.count = count;
    }

    public void compute(Double value, LocalDateTime eventTime) {
        latestValue = value;
        latestEventTime = eventTime;
        if (max.compareTo(value) < 0) {
            max = value;
        }
        if (min.compareTo(value) > 0) {
            min = value;
        }
        sum += value;
        count++;
        avg = sum / count;
        dirty.compareAndSet(false, true);
    }

    // stream|ruleId|startTime|endTime|metric|tagId
    public static String getKey(String ruleId, TimeWindow timeWindow, String metric, Integer tagId) {
        return String.join(KEY_SEPARATOR, getPrefixKey(ruleId), "" + timeWindow.getStart(), "" + timeWindow.getEnd(), metric, tagId.toString());
    }

    public static String getPrefixKey(String ruleId) {
        return String.join(KEY_SEPARATOR, KEY_PRE_NAME, ruleId);
    }

    @Override
    public String toString() {
        return "AggregateItem{" +
                "avg=" + avg +
                ", max=" + max +
                ", min=" + min +
                ", sum=" + sum +
                ", count=" + count +
                ", latestValue=" + latestValue +
                ", latestEventTime=" + latestEventTime +
                ", dirty=" + dirty +
                '}';
    }
}
